package org.apache.rocketmq.iot.client;


import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;

/**
 * @author Rocky
 * @version 1.0
 * @description
 * @email inaho00@foxmail.com
 * @createDate 2020/12/10 9:12
 * @log
 */
public class MqttDispatcherClient {
    private static MqttDispatcherClient worker = null;
    private static String topic = "/mqtt/up/a";
    private static int qos = 0;
    private static String broker = "tcp://127.0.0.1:1883";
    private static String clientId = "JavaSampleConsumer";
    private static MemoryPersistence persistence = new MemoryPersistence();

    private static Logger log = LoggerFactory.getLogger(MqttDispatcherClient.class);
    private MqttClient client;

    public MqttClient getClient() {
        return client;
    }

    private MqttDispatcherClient(String broker, String clientId, MemoryPersistence persistence, String topic, int qos){
        init(broker,clientId,persistence,topic,qos);
    }
    public static MqttDispatcherClient getInstance(){
        if (worker == null){
            worker = new MqttDispatcherClient(broker,clientId,persistence,topic,qos);
        }
        return worker;
    }

    private void init(String broker,String clientId,MemoryPersistence persistence,String topic,int qos) {
       try {
           client = new MqttClient(broker, clientId, persistence);
           MqttConnectOptions connectOptions = new MqttConnectOptions();
           connectOptions.setCleanSession(true);
           log.info("Connecting to broker: " + broker);
           client.connect(connectOptions);
           log.info("Connected");
           client.setCallback(new MqttCallback() {
               @Override public void connectionLost(Throwable throwable) {

               }

               @Override public void messageArrived(String s, MqttMessage message) throws Exception {
                   //打印从客户端到达server  再从server发送到工作线程的消息
                   byte[] payload = message.getPayload();
                   System.out.println(new String(payload, StandardCharsets.UTF_8));
               }

               @Override public void deliveryComplete(IMqttDeliveryToken token) {

               }
           });
           //这一行其实可要可不要，当个例句好了
           client.subscribe(topic, qos);
       } catch (MqttException me) {
           log.error("reason " + me.getReasonCode());
           log.error("msg " + me.getMessage());
           log.error("loc " + me.getLocalizedMessage());
           log.error("cause " + me.getCause());
           log.error("excep " + me);
           me.printStackTrace();
           System.exit(1);
       }
    }

}